package com.shujia.stream

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object CardLocationToEs {
  def main(args: Array[String]): Unit = {
    /**
     * 创建flink sql环境
     *
     */
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .inStreamingMode() //流处理模式
      //.inBatchMode()//批处理模式
      .build()

    val table: TableEnvironment = TableEnvironment.create(settings)

    /**
     * 读取卡口过车数据
     *
     */

    table.executeSql(
      """
        |CREATE TABLE cars (
        |    car STRING,
        |    city_code STRING,
        |    county_code STRING,
        |    card BIGINT,
        |    camera_id STRING,
        |    orientation STRING,
        |    road_id BIGINT,
        |    `time` BIGINT,
        |    speed DOUBLE,
        |    proc_time as PROCTIME()
        |) WITH (
        |  'connector' = 'kafka',
        |  'topic' = 'cars',
        |  'properties.bootstrap.servers' = 'master:9092',
        |  'properties.group.id' = 'testGroup',
        |  'scan.startup.mode' = 'earliest-offset',
        |  'format' = 'json'
        |)
        |
        |""".stripMargin)

    /**
     * 读取卡口维度表
     *
     */

    table.executeSql(
      """
        |
        |CREATE TABLE dim_card_mysql (
        |  card BIGINT,
        |  lon DOUBLE,
        |  lat DOUBLE,
        |  PRIMARY KEY (card) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/car',
        |   'table-name' = 'dim_card',
        |   'username' = 'root',
        |   'password' = '123456'
        |);
        |
        |""".stripMargin)

    /**
     * es sink表
     *
     */

    table.executeSql(
      """
        |CREATE TABLE cars_es (
        |    car STRING,
        |    city_code STRING,
        |    county_code STRING,
        |    card BIGINT,
        |    camera_id STRING,
        |    orientation STRING,
        |    road_id BIGINT,
        |    `time` BIGINT,
        |    speed DOUBLE,
        |    location array<DOUBLE> ,
        |    time_str STRING,
        |  PRIMARY KEY (car,`time`) NOT ENFORCED
        |) WITH (
        |  'connector' = 'elasticsearch-7',
        |  'hosts' = 'http://master:9200',
        |  'index' = 'cars'
        |)
        |
        |
        |""".stripMargin)

    /**
     * 关联卡口过车数据和卡口维度表将关联结果保存到es中
     *
     */

    table.executeSql(
      """
        |
        |insert into cars_es
        |select
        |a.car,
        |a.city_code,
        |a.county_code,
        |a.card,
        |a.camera_id,
        |a.orientation,
        |a.road_id,
        |a.`time`,
        |a.speed,
        |array[b.lon,b.lat] as location ,-- 数据位置字段
        |FROM_UNIXTIME(`time`)  as time_str --时间字段
        |
        |from
        |cars as a
        |join
        |dim_card_mysql  FOR SYSTEM_TIME AS OF a.proc_time  as b
        |on a.card=b.card
        |
        |""".stripMargin)


  }

}
